package com.ht.api.flink;

//import java.time.Duration;
//import java.util.Arrays;
//import java.util.Base64;
//import java.util.Properties;
//
//import org.apache.http.client.config.RequestConfig;
//import org.apache.http.client.methods.CloseableHttpResponse;
//import org.apache.http.client.methods.HttpPut;
//import org.apache.http.entity.StringEntity;
//import org.apache.http.impl.client.CloseableHttpClient;
//import org.apache.http.impl.client.HttpClients;
//import org.apache.http.impl.client.LaxRedirectStrategy;
//import org.apache.http.util.EntityUtils;
//import org.apache.kafka.clients.consumer.ConsumerConfig;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.apache.kafka.clients.consumer.ConsumerRecords;
//import org.apache.kafka.clients.consumer.KafkaConsumer;
//import org.apache.kafka.common.serialization.StringDeserializer;
//
//import cn.hutool.json.JSONObject;
//import cn.hutool.json.JSONUtil;
//import lombok.Data;
//import lombok.extern.slf4j.Slf4j;

//@Slf4j
public class KafkaToSelectDB {

//    @Data
//    public static class Config {
//        // Kafka配置
//        private String bootstrapServers = "192.168.0.240:9092";
//        private String topic = "mysql_data";
//        private String groupId = "selectdb-consumer";
//
//        // SelectDB配置
//        private String feHost = "selectdb-cn-rp64247ug01-public.selectdbfe.rds.aliyuncs.com";
//        private int fePort = 8080;
//        private String dbName = "test";
//        private String tableName = "users";
//        private String username = "admin";
//        private String password = "xx&*(%^Ht20241234yuiexx";
//    }
//
//    public static void main(String[] args) {
//        Config config = new Config();
//
//        // 1. Kafka消费者配置
//        Properties props = new Properties();
//        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers());
//        props.put(ConsumerConfig.GROUP_ID_CONFIG, config.getGroupId());
//        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
//        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
//
//        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
//        consumer.subscribe(Arrays.asList(config.getTopic()));
//
//        // 2. SelectDB Stream Load配置
//        String streamLoadUrl = String.format("http://%s:%d/api/%s/%s/_stream_load",
//                config.getFeHost(), config.getFePort(), config.getDbName(), config.getTableName());
//
//        try {
//            while (true) {
//                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
//
//                for (ConsumerRecord<String, String> record : records) {
//                    try {
//                        // 解析CDC JSON数据
//                        JSONObject json = JSONUtil.parseObj(record.value());
//                        JSONObject after = json.getJSONObject("after");
//                        String op = json.getStr("op");  // 使用getStr替代getString
//
//                        // 只处理插入和更新操作
//                        if ("c".equals(op) || "u".equals(op)) {
//                            // 使用HttpClient替代RestTemplate
//                            RequestConfig requestConfig = RequestConfig.custom()
//                                    .setExpectContinueEnabled(true)
//                                    .setConnectTimeout(5000)
//                                    .setSocketTimeout(5000)
//                                    .setRedirectsEnabled(true)      // 开启重定向
//                                    .setMaxRedirects(3)             // 设置最大重定向次数
//                                    .setCircularRedirectsAllowed(true)  // 允许循环重定向
//                                    .build();
//
//                            // 创建自定义的重定向策略
//                            LaxRedirectStrategy redirectStrategy = new LaxRedirectStrategy();
//
//                            try (CloseableHttpClient httpClient = HttpClients.custom()
//                                    .setDefaultRequestConfig(requestConfig)
//                                    .setRedirectStrategy(redirectStrategy)  // 设置重定向策略
//                                    .build()) {
//
//                                HttpPut httpPut = new HttpPut(streamLoadUrl);
//
//                                // 设置认证头
//                                String auth = config.getUsername() + ":" + config.getPassword();
//                                String encodedAuth = Base64.getEncoder().encodeToString(auth.getBytes());
//                                httpPut.setHeader("Authorization", "Basic " + encodedAuth);
//
//                                // 设置其他头信息
//                                httpPut.setHeader("format", "json");
//                                httpPut.setHeader("strip_outer_array", "true");
//                                httpPut.setHeader("columns", "id,name,sex,age");
//                                httpPut.setHeader("Content-Type", "application/json");
//                                httpPut.setHeader("Expect", "100-continue");
//
//                                // 构建请求体
//                                String jsonData = "[" + after.toString() + "]";
//                                StringEntity entity = new StringEntity(jsonData);
//                                entity.setChunked(true);
//                                httpPut.setEntity(entity);
//
//                                // 执行请求
//                                try (CloseableHttpResponse response = httpClient.execute(httpPut)) {
//                                    int statusCode = response.getStatusLine().getStatusCode();
//                                    String responseBody = EntityUtils.toString(response.getEntity());
//
//                                    // 添加详细日志
//                                    log.info("Initial status code: {}", statusCode);
//                                    if (response.containsHeader("Location")) {
//                                        log.info("Redirect location: {}", response.getFirstHeader("Location").getValue());
//                                    }
//                                    log.info("Stream Load response: {} - {}", statusCode, responseBody);
//
//                                    // 如果是重定向状态码
//                                    if (statusCode == 307) {
//                                        String location = response.getFirstHeader("Location").getValue();
//                                        log.info("Trying redirect to: {}", location);
//
//                                        // 创建新的请求到重定向地址
//                                        HttpPut redirectPut = new HttpPut(location);
//                                        redirectPut.setHeaders(httpPut.getAllHeaders());
//                                        redirectPut.setEntity(entity);
//
//                                        try (CloseableHttpResponse redirectResponse = httpClient.execute(redirectPut)) {
//                                            String redirectBody = EntityUtils.toString(redirectResponse.getEntity());
//                                            log.info("Redirect response: {} - {}",
//                                                    redirectResponse.getStatusLine().getStatusCode(),
//                                                    redirectBody);
//                                        }
//                                    }
//                                }
//                            }
//                        }
//                    } catch (Exception e) {
//                        log.error("处理消息失败: {}", record.value(), e);
//                    }
//                }
//            }
//        } finally {
//            consumer.close();
//        }
//    }
}
